package com.adam.example;

import com.fasterxml.jackson.databind.ObjectMapper;
import org.elasticsearch.action.bulk.*;
import org.elasticsearch.action.delete.DeleteRequestBuilder;
import org.elasticsearch.action.delete.DeleteResponse;
import org.elasticsearch.action.get.GetResponse;
import org.elasticsearch.action.get.MultiGetItemResponse;
import org.elasticsearch.action.get.MultiGetResponse;
import org.elasticsearch.action.index.IndexRequest;
import org.elasticsearch.action.index.IndexRequestBuilder;
import org.elasticsearch.action.index.IndexResponse;
import org.elasticsearch.action.update.UpdateRequest;
import org.elasticsearch.action.update.UpdateRequestBuilder;
import org.elasticsearch.action.update.UpdateResponse;
import org.elasticsearch.client.transport.TransportClient;
import org.elasticsearch.common.unit.ByteSizeUnit;
import org.elasticsearch.common.unit.ByteSizeValue;
import org.elasticsearch.common.unit.TimeValue;
import org.elasticsearch.common.xcontent.XContentBuilder;
import org.elasticsearch.common.xcontent.XContentFactory;
import org.elasticsearch.common.xcontent.XContentType;
import org.elasticsearch.script.Script;

import java.io.IOException;
import java.text.ParseException;
import java.text.SimpleDateFormat;
import java.util.concurrent.ExecutionException;

public class DocumentManagementExample {

    public static void main(String[] args) throws IOException, ParseException, ExecutionException, InterruptedException {
        TransportClient transportClient = TransportClientHolder.getInstance().getTransportClient();

        SimpleDateFormat simpleDateFormat = new SimpleDateFormat("yyyy-MM-dd");
        BooksIT booksIT = new BooksIT(7, "Java编程思想", "Java学习必读经典,殿堂级著作！赢得了全球程序员的广泛赞誉。",
                "Bruce Eckel", "java", 70.20, simpleDateFormat.parse("2007-10-01"));
        ObjectMapper objectMapper = new ObjectMapper();
        String docSource = objectMapper.writeValueAsString(booksIT);
        XContentBuilder docBuilder = XContentFactory.jsonBuilder()
                .startObject()
                .field("id", 7)
                .field("title", "Java编程思想")
                .field("description", "Java学习必读经典,殿堂级著作！赢得了全球程序员的广泛赞誉。")
                .field("author", "Bruce Eckel")
                .field("language", "java")
                .field("price", 70.20)
                .field("publish_time", "2007-10-01")
                .endObject();
        IndexResponse indexResponse = transportClient.prepareIndex("test_books", "IT", "7")
//                .setSource(docSource, XContentType.JSON)
                .setSource(docBuilder)
                .get();
        System.out.println("创建文档：" + indexResponse.status() + indexResponse.getResult()); //CREATED, CREATED
        docBuilder.close();

        GetResponse getResponse = transportClient.prepareGet("test_books", "IT", "7").get();
        System.out.println("获取文档：" + getResponse.getSourceAsString());

        UpdateRequest updateRequest = new UpdateRequest("test_books", "IT", "7");
        XContentBuilder docBuilder2 = XContentFactory.jsonBuilder().startObject()
                .field("publish_time", "2007-10-02").endObject();
        updateRequest.doc(docBuilder2);
        UpdateResponse updateResponse = transportClient.update(updateRequest).get();
        System.out.println("更新文档：" + updateResponse.status() + updateResponse.getResult()); //OK, UPDATED
        docBuilder2.close();

        UpdateRequest updateRequest2 = new UpdateRequest("test_books", "IT", "7")
                .script(new Script("ctx._source.price = 70.21"));
        UpdateResponse updateResponse2 = transportClient.update(updateRequest2).get();
        System.out.println("更新文档：" + updateResponse2.status() + updateResponse2.getResult()); //OK, UPDATED

        GetResponse getResponse2 = transportClient.prepareGet("test_books", "IT", "7").get();
        System.out.println("更新后获取文档：" + getResponse2.getSourceAsString());

        DeleteResponse deleteResponse = transportClient.prepareDelete("test_books", "IT", "7").get();
        System.out.println("删除文档：" + deleteResponse.status() + deleteResponse.getResult()); //OK, DELETED

//        BulkByScrollResponse deleteByQueryResponse = DeleteByQueryAction.INSTANCE.newRequestBuilder(transportClient)
//                .filter(QueryBuilders.termQuery("id", 7))
//                .source("test_books")
//                .get();
//        System.out.println("删除文档：" + deleteByQueryResponse.getDeleted());

        MultiGetResponse multiGetResponse = transportClient.prepareMultiGet()
                .add("test_books", "IT", "1")
                .add("test_books", "IT", "2", "3", "4")
                .get();
        MultiGetItemResponse[] multiGetItemResponses = multiGetResponse.getResponses();
        for(int i=0; i<multiGetItemResponses.length; i++) {
            GetResponse getResponse1 = multiGetItemResponses[i].getResponse();
            System.out.println("获取文档[" + (i+1) + "]：" + getResponse1.getSourceAsString());
        }

//        XContentBuilder mappingBuilder = XContentFactory.jsonBuilder()
//                .startObject()
//                .startObject("properties")
//                    .startObject("message")
//                        .field("type", "text")
//                    .endObject()
//                    .startObject("timestamp")
//                        .field("type", "date")
//                    .endObject()
//                .endObject()
//                .endObject();
//        CreateIndexResponse createIndexResponse = transportClient.admin().indices()
//                .prepareCreate("test_twitter")
//                .setSettings(Settings.builder()
//                        .put("index.number_of_shards", 1)
//                        .put("index.number_of_replicas", 0)
//                        .build())
//                .addMapping("tweet", mappingBuilder)
//                .get();
//        System.out.println("创建索引：" + createIndexResponse.isAcknowledged());
//        mappingBuilder.close();

        XContentBuilder twitterDocBuilder1 = XContentFactory.jsonBuilder()
                .startObject()
                .field("message", "tweet1")
                .field("timestamp", 1748934684)
                .endObject(),
                twitterDocBuilder2 = XContentFactory.jsonBuilder()
                        .startObject()
                        .field("message", "tweet2")
                        .field("timestamp", 1748934684)
                        .endObject(),
                twitterDocBuilder3 = XContentFactory.jsonBuilder()
                        .startObject()
                        .field("message", "tweet3")
                        .field("timestamp", 1748934684)
                        .endObject(),
                twitterDocBuilder4 = XContentFactory.jsonBuilder()
                        .startObject()
                        .field("message", "tweet3 updated")
                        .endObject();
        BulkRequestBuilder bulkRequestBuilder = transportClient.prepareBulk();
        IndexRequestBuilder indexRequestBuilder1 = transportClient.prepareIndex("test_twitter", "tweet", "1")
                .setSource(twitterDocBuilder1),
                indexRequestBuilder2 = transportClient.prepareIndex("test_twitter", "tweet", "2")
                        .setSource(twitterDocBuilder2),
                indexRequestBuilder3 = transportClient.prepareIndex("test_twitter", "tweet", "3")
                        .setSource(twitterDocBuilder3);
        UpdateRequestBuilder updateRequestBuilder = transportClient.prepareUpdate("test_twitter", "tweet", "3")
                .setDoc(twitterDocBuilder4);
        DeleteRequestBuilder deleteRequestBuilder = transportClient.prepareDelete("test_twitter", "tweet", "4");
        BulkResponse bulkResponse = bulkRequestBuilder.add(indexRequestBuilder1).add(indexRequestBuilder2).add(indexRequestBuilder3)
                .add(updateRequestBuilder).add(deleteRequestBuilder).get();
        BulkItemResponse[] bulkItemResponses = bulkResponse.getItems();
        for(int i=0; i<bulkItemResponses.length; i++) {
            BulkItemResponse bulkItemResponse = bulkItemResponses[i];
            System.out.println("Bulk请求[" + (i+1) + "]" + bulkItemResponse.status());
        }
        twitterDocBuilder1.close();
        twitterDocBuilder2.close();
        twitterDocBuilder3.close();
        twitterDocBuilder4.close();

        BulkProcessor.Listener listener = new BulkProcessor.Listener() {
            @Override
            public void beforeBulk(long executionId, BulkRequest request) {
                System.out.println("BulkProcessor.Listener beforeBulk");
            }

            @Override
            public void afterBulk(long executionId, BulkRequest request, BulkResponse response) {
                System.out.println("BulkProcessor.Listener afterBulk-success");
                BulkItemResponse[] bulkItemResponses1 = response.getItems();
                for(int i=0;i< bulkItemResponses1.length;i++) {
                    System.out.println("Bulk请求[" + (i+1) + "]:" + bulkItemResponses1[i].status());
                }
            }

            @Override
            public void afterBulk(long executionId, BulkRequest request, Throwable failure) {
                System.out.println("BulkProcessor.Listener afterBulk-exception");
            }
        };
        BulkProcessor bulkProcessor = BulkProcessor
                .builder(transportClient, listener)
                .setBulkActions(10000)
                .setBulkSize(new ByteSizeValue(20, ByteSizeUnit.MB))
                .setFlushInterval(TimeValue.timeValueSeconds(5))
                .setConcurrentRequests(5)
                .setBackoffPolicy(BackoffPolicy.exponentialBackoff(TimeValue.timeValueMillis(100), 3))
                .build();
        IndexRequest indexRequest1 = new IndexRequest("test_twitter", "tweet", "4")
                .source("{\"message\": \"bulk tweet\", \"timestamp\": 1748934684}", XContentType.JSON),
                indexRequest2 = new IndexRequest("test_twitter", "tweet", "5")
                        .source("{\"message\": \"bulk tweet\", \"timestamp\": 1748934684}", XContentType.JSON);
        UpdateRequest updateRequest3 = new UpdateRequest("test_twitter", "tweet", "4")
                .doc("{\"message\": \"bulk tweet updated\"}", XContentType.JSON);
        bulkProcessor.add(indexRequest1).add(indexRequest2).add(updateRequest3);
        bulkProcessor.flush();
        Thread.sleep(3000);  //等待异步线程执行
        bulkProcessor.close();

        transportClient.close();
    }

}
